package javaVersion.learn.streamProcess;

import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @program: myFlink
 * @description: 分离流处理
 * @author: WincoMa
 * @create: 2020-07-20 11:54
 **/
public class Pro7_SplitStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        DataStreamSource<Integer> dataStreamSource = senv.fromElements(1, 2, 3, 4, 5, 6);
        SplitStream<Integer> split = dataStreamSource.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                List<String> out = new ArrayList<>();
                if (value % 2 == 0) {
                    out.add("even");
                } else {
                    out.add("odd");
                }
                return out;
            }
        });
        //偶数流
        DataStream<Integer> even = split.select("even");
        //奇数流
        DataStream<Integer> odd = split.select("odd");
        //所有数据
        DataStream<Integer> all = split.select("odd", "even");

        odd.print();
        senv.execute();

    }
}
